package com.tl.mq.kafka_mq.config;

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.ProducerListener;

import javax.annotation.Resource;

@Configuration
public class KafkaConfig {

    @Resource
    ProducerFactory producerFactory;

    @Bean
    public KafkaTemplate<String,String> kafkaTemplate(){
        KafkaTemplate<String,String> kafkaTemplate = new KafkaTemplate<String,String>(producerFactory);
        kafkaTemplate.setProducerListener(new ProducerListener<String, String>() {
            @Override
            public void onSuccess(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata) {
                System.out.println("发送失败！！");
            }

            @Override
            public void onSuccess(String topic, Integer partition, String key, String value, RecordMetadata recordMetadata) {
                System.out.println("发送失败！！");
                System.out.println("主题：" + topic);
                System.out.println("key：" + key);
                System.out.println("value：" + value);
            }

            @Override
            public void onError(ProducerRecord<String, String> producerRecord, Exception exception) {
                System.out.println("发送失败！！");
            }

            @Override
            public void onError(String topic, Integer partition, String key, String value, Exception exception) {
                System.out.println("发送失败！！");
                System.out.println("主题：" + topic);
                System.out.println("key：" + key);
                System.out.println("value：" + value);
            }
        });
        return kafkaTemplate;
    }

}
